﻿using System;
using System.Collections.Generic;
using System.Text;
using Microsoft.Extensions.Options;
using MySqlConnector;
using System.Threading.Tasks;
using System.Threading;
using NetMicro.EventBus.Extensions;
using NetMicro.EventBus.Persistences;
using NetMicro.EventBus.Messages;
using NetMicro.Core.Extensions;

namespace NetMicro.EventBus.MySql
{
    public class DelayMessageStorage : IDelayMessageStorage
    {
        public DelayMessageStorage(IOptionsMonitor<EventBusMySqlOptions> options)
        {
            Options = options;
        }

        private IOptionsMonitor<EventBusMySqlOptions> Options { get; }

        /// <summary>
        /// 保存延迟消息
        /// </summary>
        /// <param name="message"></param>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public async Task<bool> SaveAsync(MessageStorageTransfer message,
          CancellationToken cancellationToken = default)
        {
            var sql = $@"
INSERT INTO `{Options.CurrentValue.DelayTableName}`
(`Id`,`EventBody`, `CreateTime`, `DelayAt`, `ExpireTime`, `RetryCount`, `IsLocking`, `LockEnd`)
VALUES(@Id,@EventBody, @CreateTime, @DelayAt, @ExpireTime, @RetryCount, @IsLocking, @LockEnd);
";
            var parameters = new[]
            {
                new MySqlParameter("@Id", MySqlDbType.Int64) {Value = message.Id},
                new MySqlParameter("@EventBody", MySqlDbType.VarChar) {Value = message.EventBody},
                new MySqlParameter("@CreateTime", MySqlDbType.Int64) {Value = message.CreateTime},
                new MySqlParameter("@DelayAt", MySqlDbType.Int64) {Value = message.DelayAt},
                new MySqlParameter("@ExpireTime", MySqlDbType.Int64) {Value = Options.CurrentValue.DelayMessageExpireTime.GetTimestampSeconds()},
                new MySqlParameter("@RetryCount", MySqlDbType.VarChar) {Value = message.RetryCount},
                new MySqlParameter("@IsLocking", MySqlDbType.Bit) {Value = message.IsLocking},
                new MySqlParameter("@LockEnd", MySqlDbType.Int64) {Value = message.LockEnd}
            };
            await using var connection = new MySqlConnection(Options.CurrentValue.ConnectionString);
            return connection.ExecuteNonQuery(sql, sqlParams: parameters) > 0;
        }

        /// <summary>
        /// 延迟消息添加锁
        /// </summary>
        /// <param name="id"></param>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public async Task<bool> LockAsync(long id,
         CancellationToken cancellationToken = default)
        {
            var sql = $@"
UPDATE `{Options.CurrentValue.DelayTableName}`
SET `IsLocking`=1,`LockEnd`={Options.CurrentValue.MessageLockExpireTime.GetTimestampSeconds()}
WHERE `Id` = @Id
";
            object[] sqlParams =
            {
                new MySqlParameter("@Id", id)
            };
            await using var connection = new MySqlConnection(Options.CurrentValue.ConnectionString);
            return connection.ExecuteNonQuery(sql, sqlParams: sqlParams) > 0;
        }

        /// <summary>
        /// 延迟消息释放锁
        /// </summary>
        /// <param name="id"></param>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public async Task<bool> UnLockAsync(long id,
         CancellationToken cancellationToken = default)
        {
            var sql = $@"
UPDATE `{Options.CurrentValue.DelayTableName}`
SET `IsLocking`=0,`LockEnd`=0
WHERE `Id` = @Id
";
            object[] sqlParams =
            {
                new MySqlParameter("@Id", id)
            };
            await using var connection = new MySqlConnection(Options.CurrentValue.ConnectionString);
            return connection.ExecuteNonQuery(sql, sqlParams: sqlParams) > 0;
        }

        /// <summary>
        /// 延迟消息锁到期之后未释放的锁,释放锁
        /// </summary>
        /// <param name="id"></param>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public async Task<bool> ExpireTimeUnLockAsync(
         CancellationToken cancellationToken = default)
        {
            var sql = $@"
UPDATE `{Options.CurrentValue.DelayTableName}`
SET `IsLocking`=0,`LockEnd`=0
WHERE `LockEnd` < {TimeSpan.Zero.GetTimestampSeconds()} AND `IsLocking`=1
";
            await using var connection = new MySqlConnection(Options.CurrentValue.ConnectionString);
            return connection.ExecuteNonQuery(sql) > 0;
        }
        /// <summary>
        /// 删除完成延迟消息
        /// </summary>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public async Task<bool> DeleteMessageAsync(long id, CancellationToken cancellationToken = default)
        {
            var sql = $@"DELETE FROM `{Options.CurrentValue.DelayTableName}` WHERE Id=@Id;";
            object[] sqlParams =
           {
                new MySqlParameter("@Id", id)
            };
            await using var connection = new MySqlConnection(Options.CurrentValue.ConnectionString);
            return connection.ExecuteNonQuery(sql, sqlParams: sqlParams) > 0;
        }

        /// <summary>
        /// 获取延迟消息
        /// </summary>
        /// <returns></returns>
        public async Task<IEnumerable<MessageStorageTransfer>> GetMessagesAsync()
        {
            var sql =
                        $"SELECT `Id`,`EventBody`,`CreateTime`,`DelayAt`,`ExpireTime`,`RetryCount` FROM `{Options.CurrentValue.DelayTableName}` WHERE `IsLocking`=0 ORDER BY `DelayAt` LIMIT 5;";
            await using var connection = new MySqlConnection(Options.CurrentValue.ConnectionString);
            var result = connection.ExecuteReader(sql, reader =>
            {
                var messages = new List<MessageStorageTransfer>();
                while (reader.Read())
                {
                    messages.Add(new MessageStorageTransfer
                    {
                        Id = reader.GetInt64(0),
                        EventBody = reader.GetString(1),
                        CreateTime = reader.GetInt64(2),
                        DelayAt = reader.GetInt64(3),
                        ExpireTime = reader.GetInt64(4),
                        RetryCount = reader.GetInt32(5),
                    });
                }
                return messages;
            });
            return result;
        }

        /// <summary>
        /// 删除到期消息
        /// </summary>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public async Task<bool> DeleteExpiresAsync(CancellationToken cancellationToken = default)
        {
            var now = TimeSpan.Zero.GetTimestampSeconds();
            var sql = $@"DELETE FROM `{Options.CurrentValue.DelayTableName}` WHERE `ExpireTime` > 0 AND `ExpireTime` < {now};";
            await using var connection = new MySqlConnection(Options.CurrentValue.ConnectionString);
            return connection.ExecuteNonQuery(sql) > 0;
        }
    }
}
